package event

import (
	"context"
	"fmt"
	"github.com/ethereum/go-ethereum/common/mclock"
	"sync"
	"time"
)

type Subscription interface {
	Err() <- chan error  // returns the error channel
	Unsubscribe()   	 // cancels sending of events , closing the error channel
}

/*该函数实际上dui producer进行了封装*/
func NewSubscription(producer func(quitchan <- chan struct{}) error) Subscription{
	s := &funcSub{ unsub:make(chan struct{}) , err : make(chan error,1)}
	go func (){
		defer close(s.err)
		/*执行producer函数 将s.unsub加入到producer里*/
		err := producer(s.unsub)

		s.mu.Lock()
		defer s.mu.Unlock()

		if !s.unsubscribed{
			if err != nil{
				fmt.Println("发送err")
				s.err <- err
			}
			fmt.Println("进行取消订阅")
			s.unsubscribed = true
		}
	}()
	return s
}

type funcSub struct {
	unsub chan struct{}
	err  chan error
	mu sync.Mutex
	unsubscribed bool
}

func ( s * funcSub)Unsubscribe() {
	s.mu.Lock()
	if s.unsubscribed{
		fmt.Println("直接退出")
		s.mu.Unlock()
		return
	}

	s.unsubscribed = true
	fmt.Println("关闭订阅")
	close(s.unsub)
	s.mu.Unlock()

	/*wait for producer shutdown*/
	<- s.err
}

func (s * funcSub)Err() <- chan error{
	return s.err
}

func Resubscribe(backoffMax time.Duration , fn ResubscribeFunc) Subscription{
	s := &resubscribeSub{
		waitTime:   backoffMax / 10,
		backoffMax: backoffMax,
		fn:         fn,
		err:        make(chan error),
		unsub:      make(chan struct{}),
	}
	go s.loop()
	return s
}

type ResubscribeFunc func(ctx context.Context) (Subscription ,error)

type resubscribeSub struct {
	fn ResubscribeFunc
	err chan error
	unsub  chan struct{}
	unsubOnce sync.Once
	lastTry mclock.AbsTime
	waitTime, backoffMax time.Duration
}

func (s *resubscribeSub) Unsubscribe() {
	s.unsubOnce.Do(func() {
		s.unsub <- struct{}{}
		<-s.err
	})
}

func (s *resubscribeSub) Err() <-chan error {
	return s.err
}

func (s * resubscribeSub) loop (){
	defer close(s.err)

	var done bool
	for !done {
		sub := s.subscribe()
		if sub == nil{
			break
		}

		done = s.waitForError(sub)  //如果发生错误则返回false  代笔：要重新订阅
		sub.Unsubscribe()
	}
}

func (s * resubscribeSub)  subscribe() Subscription {
	subscribed := make(chan error)
	var sub Subscription
	count :=0
	for   {
		count ++
		fmt.Printf("第%d次进行重新订阅................",count)
		s.lastTry = mclock.Now()
		ctx ,cancle := context.WithCancel(context.Background())
		go func() {
			rsub , err := s.fn(ctx)
			sub = rsub
			subscribed <- err
		}()

		select {
		case err := <- subscribed :
			cancle()
			if err == nil{
				if sub == nil{
					panic("event : ResubscribeFunc returned nil subscribtion and no error ")
				}

				return sub
			}else if err!= nil{
				fmt.Println("发生错误:",err)
			}

		/*如果err!=nil
		则要进行重新订阅 ，计算间隔时间 */
			//if s.backoffWait(){ //不再等待
			//	return nil // unsubscribed during wait
			//}
			time.Sleep(time.Second)

		case <- s.unsub:
			cancle()
			<- subscribed
			return nil

		}
	}
		return nil
}

func (s * resubscribeSub)waitForError(sub Subscription)bool {
	defer sub.Unsubscribe()
	select {
	case err := <- sub.Err():
		return err == nil
	case <- s.unsub:
		return true
	}
}

/*
Resubscribe每隔backoff 时间调用一次fn
调用fn的间隔时间根据错误率进行调整，但不会超过backoffMax。
*/
func (s *resubscribeSub) backoffWait() bool {
	if time.Duration(mclock.Now()-s.lastTry ) >s.backoffMax{
		s.waitTime =s.backoffMax/10
	}else{
		s.waitTime *=2
		if s.waitTime > s.backoffMax {
			s.waitTime = s.backoffMax
		}
	}

	t := time.NewTicker(s.waitTime)
	defer t.Stop()
	select {
	case <- t.C:
		return false
	case <- s.unsub:
		return true

	}
}

/* 集中订阅控制器 */
//SubscriptionScope提供了一种一次取消订阅多个订阅的功能。
type SubscriptionScope struct {
	mu sync.Mutex
	subs map[*scopeSub]struct{}  //使用map是因为哈希表比链表更好删除
	closed bool
}

type scopeSub struct {
	sc *SubscriptionScope
	s Subscription
}


func (sc * SubscriptionScope)Track(s Subscription) Subscription{
	sc.mu.Lock()
	defer sc.mu.Unlock()
	if sc.closed{
		return nil
	}
	if sc.subs == nil{
		sc.subs = make(map[*scopeSub]struct{})
	}
	ss := &scopeSub{sc:sc,s:s}
	//加入该订阅
	sc.subs[ss] = struct{}{}
	return ss
}

func (sc * SubscriptionScope)Close() {
	sc.mu.Lock()
	defer sc.mu.Unlock()
	if sc.closed {
		return
	}
	sc.closed = true
	for s:= range sc.subs{
		s.s.Unsubscribe()
	}
	sc.subs = nil
}

func (s * scopeSub)Unsubscribe(){
	s.s.Unsubscribe()
	s.sc.mu.Lock()
	defer s.sc.mu.Unlock()
	delete(s.sc.subs,s)
}


func (s * scopeSub)Err() <-chan error{
	return s.s.Err()
}